package felix.rule.service.impl;

import felix.actor.core.DefaultTActorSystem;
import felix.actor.core.TActorRef;
import felix.actor.core.TActorSystem;
import felix.actor.core.TActorSystemSettings;
import felix.rule.data.plugin.ComponentLifecycleEvent;
import felix.rule.data.plugin.ComponentLifecycleMsg;
import felix.rule.service.ActorService;
import felix.rule.service.actors.ActorSystemContext;
import felix.rule.service.actors.app.AppActor;
import felix.rule.service.actors.app.AppInitMsg;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.core.env.Environment;
import org.springframework.core.env.Profiles;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service
@Slf4j
public class DefaultActorService implements ActorService {

    public static final String APP_DISPATCHER_NAME = "app-dispatcher";
    public static final String RULE_DISPATCHER_NAME = "rule-dispatcher";
    public static final String ENGINE_DISPATCHER_NAME = "engine-dispatcher";

    @Autowired
    private ActorSystemContext actorContext;

    private TActorSystem system;

    private TActorRef appActor;

    @Value("${actors.system.throughput:5}")
    private int actorThroughput;

    @Value("${actors.system.max_actor_init_attempts:10}")
    private int maxActorInitAttempts;

    @Value("${actors.system.scheduler_pool_size:1}")
    private int schedulerPoolSize;

    @Value("${actors.system.app_dispatcher_pool_size:1}")
    private int appDispatcherSize;

    @Value("${actors.system.rule_dispatcher_pool_size:2}")
    private int ruleDispatcherSize;

    @Value("${actors.system.engine_dispatcher_pool_size:4}")
    private int engineDispatcherSize;

    @Autowired
    private Environment environment;

    private boolean isInstall() {
        return environment.acceptsProfiles(Profiles.of("install"));
    }

    @PostConstruct
    public void initActorSystem() {
        if (!isInstall()) {
            log.info("Initializing actor system.");
            actorContext.setActorService(this);
            TActorSystemSettings settings = new TActorSystemSettings(actorThroughput, schedulerPoolSize, maxActorInitAttempts);
            system = new DefaultTActorSystem(settings);

            system.createDispatcher(APP_DISPATCHER_NAME, initDispatcherExecutor(APP_DISPATCHER_NAME, appDispatcherSize));
            system.createDispatcher(RULE_DISPATCHER_NAME, initDispatcherExecutor(RULE_DISPATCHER_NAME, ruleDispatcherSize));
            system.createDispatcher(ENGINE_DISPATCHER_NAME, initDispatcherExecutor(ENGINE_DISPATCHER_NAME, engineDispatcherSize));

            actorContext.setActorSystem(system);

            appActor = system.createRootActor(APP_DISPATCHER_NAME, new AppActor.ActorCreator(actorContext));
            actorContext.setAppActor(appActor);

            log.info("Actor system initialized.");
        }
    }

    private ExecutorService initDispatcherExecutor(String dispatcherName, int poolSize) {
        if (poolSize == 0) {
            int cores = Runtime.getRuntime().availableProcessors();
            poolSize = Math.max(1, cores / 2);
        }
        if (poolSize == 1) {
            return Executors.newSingleThreadExecutor();
        } else {
            return Executors.newWorkStealingPool(poolSize);
        }
    }

    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        if (!isInstall()) {
            log.info("Received application ready event. Sending application init message to actor system");
            appActor.tellWithHighPriority(new AppInitMsg());
        }
    }

    @PreDestroy
    public void stopActorSystem() {
        if (system != null) {
            log.info("Stopping actor system.");
            system.stop();
            log.info("Actor system stopped.");
        }
    }


    @Override
    public void onEntityStateChange(String ruleId, String type, String entityId, ComponentLifecycleEvent state) {
        log.trace("[{}] Processing {} state change event: {}", ruleId, entityId, state);
        broadcast(new ComponentLifecycleMsg(ruleId, type, entityId, state));
    }

    private void broadcast(ComponentLifecycleMsg msg) {
        appActor.tell(msg);
    }

}
